package com.syf.kit.soa.core.send.protocol.client;

import com.syf.kit.soa.core.SyfRpcMsg;
import com.syf.kit.soa.core.send.protocol.TcpClient;
import com.syf.kit.soa.core.send.protocol.handle.RpcDataDecoder;
import com.syf.kit.soa.core.send.protocol.handle.RpcDataEncoder;
import com.syf.kit.soa.core.send.protocol.handle.RpcKeepAliveSendHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.AttributeKey;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.concurrent.TimeUnit;

/**
 * 池化连接，内部包含连接池
 */
public class PooledTcpClient implements TcpClient {
    private String host;

    private Integer port;

    private String magicNum;

    private long connTimeout;

    /**
     * 连接池最大数
     */
    private int poolMaxSize;
    /**
     * 连接池保留空闲连接的最低限制
     */
    private int poolMinSize;

    private GenericObjectPool<LongTcpConnection> connectPool;

    public PooledTcpClient(String host, int port, String magicNum, long connTimeout, int poolMaxSize, int poolMinSize) {
        this(host, port, magicNum, connTimeout, poolMaxSize, poolMinSize, 60L);
    }

    public PooledTcpClient(String host, int port, String magicNum, long connTimeout, int poolMaxSize, int poolMinSize, long beatInterval) {
        this.host = host;
        this.port = port;
        this.magicNum = magicNum;
        this.connTimeout = connTimeout;
        this.poolMaxSize = poolMaxSize;
        this.poolMinSize = poolMinSize;

        initialize(beatInterval);
    }

    private void initialize(long beatInterval) {
        final AttributeKey<Boolean> CONN_USABLE_KEY = AttributeKey.valueOf("CONN_IS_USABLE");

        // 初始化netty启动器
        Bootstrap bootstrap = new Bootstrap();
        EventLoopGroup loopGroup = new NioEventLoopGroup();
        bootstrap.group(loopGroup).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true).handler(new ChannelInitializer<Channel>() {
            @Override
            protected void initChannel(Channel channel) throws Exception {
                ChannelPipeline pipeline = channel.pipeline();
                pipeline.addLast(new IdleStateHandler(0L, 0L, beatInterval, TimeUnit.MILLISECONDS));
                pipeline.addLast(new RpcDataDecoder());
                pipeline.addLast(new RpcDataEncoder());
                pipeline.addLast(new RpcKeepAliveSendHandler(magicNum, CONN_USABLE_KEY));
            }
        });

        // 初始化connection factory
        GenericObjectPoolConfig<LongTcpConnection> config = new GenericObjectPoolConfig<>();
        config.setMinIdle(poolMinSize);
        config.setMaxTotal(poolMaxSize);
        config.setMaxIdle(poolMaxSize);
        config.setTestOnBorrow(true);
        config.setTestOnCreate(false);
        config.setTestOnReturn(false);
        config.setTestWhileIdle(true);
        /**
         * 指定了在两次对象回收（eviction）运行之间的时间间隔，设置10分钟
         * */
        config.setTimeBetweenEvictionRuns(Duration.of(600, ChronoUnit.SECONDS));
        /**
         * 对象最小的空闲时间，在对象回收运行时，如果一个连接超过此时间，且超过maxIdle，则会被回收.设置6分钟
         * */
        config.setMinEvictableIdleTime(Duration.of(360, ChronoUnit.SECONDS));
        config.setNumTestsPerEvictionRun(5);

        connectPool = new GenericObjectPool<>(new LongTcpConnectionFactory(host, port, bootstrap, connTimeout, CONN_USABLE_KEY), config);
    }

    @Override
    public SyfRpcMsg send(SyfRpcMsg reqMsg, long reqTimeout) {
        try {
            LongTcpConnection longTcpConnection = connectPool.borrowObject(connTimeout);
            return longTcpConnection.send(reqMsg, reqTimeout);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void close() {
        connectPool.close();
    }
}
